package demo.dangdang.elastic;

import javax.sql.DataSource;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobRootConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration.Builder;
import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;

import demo.dangdang.elastic.job.DataflowJobDemo;
import demo.dangdang.elastic.job.SimpleJobDemo;
import demo.dangdang.elastic.job.listener.ElasticJobListenerDemo;

/**
 * 
 * elastic-job是当当开源的一款非常好用的作业框架，在这之前，我们开发定时任务一般都是使用quartz或者spring-task（ScheduledExecutorService），
 * 无论是使用quartz还是spring-task，我们都会至少遇到两个痛点：
 * <li>1.不敢轻易跟着应用服务多节点部署，可能会重复多次执行而引发系统逻辑的错误。
 * <li>2.quartz的集群仅仅只是用来HA，节点数量的增加并不能给我们的每次执行效率带来提升，即不能实现水平扩展。
 * <p>
 * elastic底层的任务调度还是使用的quartz，通过zookeeper来动态给job节点分片。
 * <h1>弹性分布式实现</h1>
 * <li>第一台服务器上线触发主服务器选举。主服务器一旦下线，则重新触发选举，选举过程中阻塞，只有主服务器选举完成，才会执行其他任务。
 * <li>某作业服务器上线时会自动将服务器信息注册到注册中心，下线时会自动更新服务器状态。
 * <li>主节点选举，服务器上下线，分片总数变更均更新重新分片标记。
 * <li>定时任务触发时，如需重新分片，则通过主服务器分片，分片过程中阻塞，分片结束后才可执行任务。如分片过程中主服务器下线，则先选举主服务器，再分片。
 * <li>通过上一项说明可知，为了维持作业运行时的稳定性，运行过程中只会标记分片状态，不会重新分片。分片仅可能发生在下次任务触发前。
 * <li>每次分片都会按服务器IP排序，保证分片结果不会产生较大波动。
 * <li>实现失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。
 * <p>
 * <h3>很大体量的用户需要在特定的时间段内计息完成</h3>
 * 我们肯定是希望我们的任务可以通过集群达到水平扩展，集群里的每个节点都处理部分用户，不管用户数量有多庞大，我们只要增加机器就可以了。
 * 比如单台机器特定时间能处理n个用户，2台机器处理2n个用户，3台3n，4台4n...，再多的用户也不怕了。
 * 使用elastic-job开发的作业都是zookeeper的客户端，比如我希望3台机器跑job，我们将任务分成3片，框架通过zk的协调，最终会让3台机器分别分配到0,1,2的任务片。
 * 比如server0-->0，server1-->1，server2-->2，当server0执行时，可以只查询id%3==0的用户，server1执行时，只查询id%3==1的用户，server2执行时，只查询id%3==2的用户。
 * 
 * <h3>任务部署多节点引发重复执行</h3><br>
 * 在上面的基础上，我们再增加server3，此时，server3分不到任务分片，因为只有3片，已经分完了。没有分到任务分片的作业程序将不执行。
 * 如果此时server2挂了，那么server2的分片项会分配给server3，server3有了分片，就会替代server2执行。
 * 如果此时server3也挂了，只剩下server0和server1了，框架也会自动把server3的分片随机分配给server0或者server1，可能会这样，server0-->0，server1-->1,2。
 * 这种特性称之为弹性扩容，即elastic-job名称的由来。
 * 
 * <p>
 * elastic-job是不支持单机多实例的，通过zk的协调分片是以ip为单元的。
 * 
 * 
 */
public class ElasticJobDemo {

    /**
     * 注册中心配置。 用于注册和协调作业分布式行为的组件，目前仅支持Zookeeper
     * 
     * @return
     */
    public static ZookeeperConfiguration zkConfig() {
        // 定义Zookeeper注册中心配置对象,连接Zookeeper服务器的列表。 多个地址用逗号分隔如: host1:2181,host2:2181
        ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("localhost:2181", "namespace1");
        // 最大重试次数
        zkConfig.setMaxRetries(3);
        // 等待重试的间隔时间的初始值
        zkConfig.setBaseSleepTimeMilliseconds(1_000);
        // 等待重试的间隔时间的最大值
        zkConfig.setMaxSleepTimeMilliseconds(3_000);
        // 会话超时时间
        zkConfig.setSessionTimeoutMilliseconds(60_000);
        // 连接超时时间
        zkConfig.setConnectionTimeoutMilliseconds(15_000);
        // 连接Zookeeper的权限令牌， 缺省为不需要权限验证
        zkConfig.setDigest(null);
        return zkConfig;
    }

    /**
     * 定义日志数据库事件溯源配置
     * 
     * @param dataSource
     * @return
     */
    public static JobEventConfiguration jobEventCfg(DataSource dataSource) {
        JobEventConfiguration cfg = new JobEventRdbConfiguration(dataSource);
        return cfg;
    }

    public void demo1(ZookeeperConfiguration zkConfig, DataSource dataSource) {
        // 定义Zookeeper注册中心
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);

        JobCoreConfiguration.Builder builder = JobCoreConfiguration.newBuilder("oneOffElasticDemoJob", "0/5 * * * * ?",
                1);
        builder.jobProperties("", "");
        JobCoreConfiguration jobConfig1 = builder.build();
        // 定义作业1配置对象

        // 连接注册中心
        regCenter.init();

        ElasticJobListenerDemo listener = new ElasticJobListenerDemo();

        SimpleJobConfiguration simpleJobCfg = simpleJobCfg();

        LiteJobConfiguration liteJobCfg = liteJobCfg(simpleJobCfg);

        JobEventConfiguration jobEventCfg = jobEventCfg(dataSource);

        JobScheduler jobScheduler = new JobScheduler(regCenter, liteJobCfg, jobEventCfg, listener);
        jobScheduler.init();

        // JobScheduleController controller = new JobScheduleController(jobScheduler,
        // jobDetail, schedulerFacade,
        // triggerIdentity);
        // controller.scheduleJob("");

    }

    /**
     * Lite作业配置
     * 
     * @param jobConfig
     * @return
     */
    public static LiteJobConfiguration liteJobCfg(JobTypeConfiguration jobConfig) {
        Builder builder = LiteJobConfiguration.newBuilder(jobConfig)
                /**
                 * 监控作业运行时状态
                 * <li>每次作业执行时间和间隔时间均非常短的情况，建议不监控作业运行时状态以提升效率。因为是瞬时状态，所以无必要监控。请用户自行增加数据堆积监控。并且不能保证数据重复选取，应在作业中实现幂等性。
                 * <li>每次作业执行时间和间隔时间均较长的情况，建议监控作业运行时状态，可保证数据不会重复选取。
                 */
                .monitorExecution(true)
                // 作业监控端口： 建议配置作业监控端口, 方便开发者dump作业信息。使用方法: echo “dump” | nc 127.0.0.1 9888
                .monitorPort(-1)
                // 最大允许的本机与注册中心的时间误差秒数。 如果时间误差超过配置秒数则作业启动时将抛异常。 配置为-1表示不校验时间误差
                .maxTimeDiffSeconds(-1)
                // 作业分片策略实现类全路径. 默认使用平均分配策略
                .jobShardingStrategyClass("-1")
                // 修复作业服务器不一致状态服务调度间隔时间，配置为小于1的任意值表示不执行修复.单位：分钟
                .reconcileIntervalMinutes(10);

        LiteJobConfiguration liteJobConfig = builder.build();
        return liteJobConfig;
    }

    /**
     * 作业配置分为3级，分别是JobCoreConfiguration，JobTypeConfiguration和LiteJobConfiguration。
     * LiteJobConfiguration使用JobTypeConfiguration，JobTypeConfiguration使用JobCoreConfiguration，层层嵌套。
     * 
     * JobTypeConfiguration根据不同实现类型分为SimpleJobConfiguration，DataflowJobConfiguration和ScriptJobConfiguration。
     * 
     * @return
     */
    public static SimpleJobConfiguration simpleJobCfg() {
        // 定义作业核心配置
        JobCoreConfiguration jobCoreCfg = JobCoreConfiguration.newBuilder("demoSimpleJob", "0/15 * * * * ?", 10)
                // 描述
                .description("simple job描述信息")
                // 是否开启任务执行失效转移，只有对monitorExecution的情况下才可以开启失效转移。开启表示如果作业在一次任务执行中途宕机，允许将该次未完成的任务在另一作业节点上补偿执行
                .failover(false)
                // 是否开启错过任务重新执行
                .misfire(true)
                // 分片序列号和参数用等号分隔，多个键值对用逗号分隔。 分片序列号从0开始，不可大于或等于作业分片总数 如：0=a,1=b,2=c
                .shardingItemParameters("0=a,1=b,2=c")
                // 作业自定义参数，可通过传递该参数为作业调度的业务方法传参，用于实现带参数的作业。 例：每次获取的数据量、作业实例从数据库读取的主键等
                .jobParameter("")
                /**
                 * 设置作业属性 配置jobProperties定义的枚举控制Elastic-Job的实现细节,
                 * <li>JOB_EXCEPTION_HANDLER用于扩展异常处理类,
                 * <li>EXECUTOR_SERVICE_HANDLER用于扩展作业处理线程池类
                 */
                .jobProperties("key", "value")
                // 构建作业配置对象
                .build();
        // 定义SIMPLE类型配置
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(jobCoreCfg,
                SimpleJobDemo.class.getCanonicalName());
        return simpleJobConfig;

    }

    public void dataflowJobCfg() {
        // 定义作业核心配置
        JobCoreConfiguration dataflowCoreConfig = JobCoreConfiguration
                .newBuilder("demoDataflowJob", "0/30 * * * * ?", 10).build();

        /**
         * 是否流式处理数据
         * <li>如果流式处理数据, 则fetchData不返回空结果将持续执行作业
         * <li>如果非流式处理数据, 则处理数据完成后作业结束
         */
        boolean streamingProcess = false;

        // 定义DATAFLOW类型配置
        DataflowJobConfiguration dataflowJobConfig = new DataflowJobConfiguration(dataflowCoreConfig,
                DataflowJobDemo.class.getCanonicalName(), streamingProcess);

        // 定义Lite作业根配置
        JobRootConfiguration dataflowJobRootConfig = LiteJobConfiguration.newBuilder(dataflowJobConfig).build();
    }

    public void scriptJobCfg() {

        // 定义作业核心配置配置
        JobCoreConfiguration scriptCoreConfig = JobCoreConfiguration.newBuilder("demoScriptJob", "0/45 * * * * ?", 10)
                .build();
        // 定义SCRIPT类型配置
        ScriptJobConfiguration scriptJobConfig = new ScriptJobConfiguration(scriptCoreConfig, "test.sh");
        // 定义Lite作业根配置
        JobRootConfiguration scriptJobRootConfig = LiteJobConfiguration.newBuilder(scriptJobConfig).build();

    }

}
